package org.lql.chaos.rocketmq.controller;

import lombok.extern.slf4j.Slf4j;
import org.lql.chaos.rocketmq.service.consumer.ConsumerMonitorService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import java.util.Set;

/**
 * @author: liangqinglong
 * @date: 2025-08-17 00:40
 * @description: 消息消费者管理控制器
 **/
@Slf4j
@RestController
@RequestMapping("/consumer")
public class MessageConsumeController {

    @Autowired
    private ConsumerMonitorService consumerMonitorService;

    /**
     * 启动消费者
     *
     * @param consumerGroup 消费者组
     * @param topic         主题
     * @param tag           标签
     * @return 结果信息
     */
    @PostMapping("/start")
    public String startConsumer(@RequestParam String consumerGroup,
                                @RequestParam String topic,
                                @RequestParam String tag) {
        try {
            consumerMonitorService.createAndStartConsumer(consumerGroup, topic, tag);
            return "消费者启动成功: " + consumerGroup;
        } catch (Exception e) {
            log.error("启动消费者失败", e);
            return "消费者启动失败: " + e.getMessage();
        }
    }

    /**
     * 停止消费者
     *
     * @param consumerGroup 消费者组
     * @return 结果信息
     */
    @PostMapping("/stop")
    public String stopConsumer(@RequestParam String consumerGroup) {
        try {
            consumerMonitorService.shutdownConsumer(consumerGroup);
            return "消费者停止成功: " + consumerGroup;
        } catch (Exception e) {
            log.error("停止消费者失败", e);
            return "消费者停止失败: " + e.getMessage();
        }
    }

    /**
     * 获取所有消费者状态
     *
     * @return 消费者组列表
     */
    @GetMapping("/list")
    public Set<String> listConsumers() {
        return consumerMonitorService.getConsumers().keySet();
    }
}